package com.xiangxiao.rpan.data.listener;

import com.alibaba.fastjson.JSONObject;
import com.xiangxiao.rpan.data.channel.GreetingsStreams;
import com.xiangxiao.rpan.data.entity.KafkaAlarm;
import com.xiangxiao.rpan.data.model.Greetings;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * @auther xiangxiao
 * @email 573768011@qq.com
 * @data 2024/5/30 17:12
 */
@Component
@Slf4j
public class GreetingsListener {
  @StreamListener(GreetingsStreams.INPUT)
  public void handleGreetings(Message<?> message) {
    log.info("Received greetings: {}", message.getPayload());
    Object payload = message.getPayload();
    Greetings greetings = JSONObject.parseObject(String.valueOf(payload), Greetings.class);
    Acknowledgment ack = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
    if (ack != null) {
      ack.acknowledge();
    };
  }
}
